package com.hbase.zhou.telecom;

import com.hbase.zhou.db.HBaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.junit.Test;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @Description: 模拟电信通话业务
 * @Author: ZhOu
 * @Date: 2018/3/20
 */
public class TelephoneTest {

    /**
     * 创建记录表
     */
    @Test
    public void createTable() {
        List<HColumnDescriptor> list = new ArrayList<>();
        HColumnDescriptor descriptor = new HColumnDescriptor("cf1".getBytes());
        descriptor.setMinVersions(1);
        descriptor.setMaxVersions(5);
        //大小在1K到16M之间，默认是64K，这里设为128K
        descriptor.setBlocksize(131072);
        descriptor.setInMemory(true);
        list.add(descriptor);
        list.add(new HColumnDescriptor("cf2".getBytes()));
        boolean flag = HBaseUtil.createTable("tel_db", list);
        if (flag) {
            System.out.println("表创建成功");
        } else {
            System.out.println("表创建失败");
        }
    }

    /**
     * 生成随机数，模拟通话记录
     * 记录包括：
     * 行键：号码_日期
     * 通话者电话
     * 通话时长
     * 通话类型：0-主叫 1-被叫
     */
    @Test
    public void putData() {
        List<Put> list = new ArrayList<>();
        int count = 10;
        for (int i = 0; i < count; i++) {
            String phoneNum = "186" + (new Random().nextInt(89999999) + 10000000);
            int view = 100;
            for (int j = 0; j < view; j++) {
                String date = getDate();
                String rowKey = phoneNum + "_" + date;
                Put put = new Put(rowKey.getBytes());
                put.addColumn("cf1".getBytes(), "viber".getBytes(), ("186" + (new Random().nextInt(89999999) + 10000000)).getBytes());
                put.addColumn("cf1".getBytes(), "type".getBytes(), (new Random().nextBoolean() ? "0" : "1").getBytes());
                put.addColumn("cf1".getBytes(), "duration".getBytes(), (new Random().nextInt(100) + "").getBytes());
                put.addColumn("cf1".getBytes(), "datetime".getBytes(), date.getBytes());
                list.add(put);
            }
        }

        boolean flag = HBaseUtil.putData("tel_db", list);
        if (flag) {
            System.out.println("插入成功");
        } else {
            System.out.println("插入失败");
        }
    }

    /**
     * 使用前缀过滤器，查询指定号码的通话记录
     */
    @Test
    public void getData() {
        PrefixFilter filter = new PrefixFilter("18694342830_".getBytes());
        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.setReversed(true);
        try {
            Table table = HBaseUtil.getConnection().getTable(TableName.valueOf("tel_db"));
            ResultScanner results = table.getScanner(scan);
            for (Result result : results) {
                for (Cell cell : result.rawCells()) {
                    System.out.println("rowkey=" + new String(CellUtil.cloneRow(cell))
                            + "\t family=" + new String(CellUtil.cloneFamily(cell))
                            + "\t qualifier=" + new String(CellUtil.cloneQualifier(cell))
                            + "\t value=" + new String(CellUtil.cloneValue(cell))
                    );
                }
            }
            results.close();
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    private String getDate() {
        LocalDateTime localDateTime = LocalDateTime.of(2017, 1, 1, 0, 0, 0);
        localDateTime = localDateTime.plusSeconds(new Random().nextInt(31536000));
        return localDateTime.format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
    }

}
